package cn.ac.iie.di.ocr.mpp;

import cn.ac.iie.di.ocr.util.GlobalParameters;
import com.datastax.driver.core.*;
import com.datastax.driver.core.exceptions.InvalidQueryException;
import com.datastax.driver.core.exceptions.NoHostAvailableException;
import com.datastax.driver.core.exceptions.OperationTimedOutException;
import com.datastax.driver.core.exceptions.QueryExecutionException;
import com.datastax.driver.core.policies.DCAwareRoundRobinPolicy;
import com.datastax.driver.core.policies.DefaultRetryPolicy;
import org.apache.commons.codec.DecoderException;
import org.apache.commons.codec.binary.Hex;
import org.codehaus.jackson.map.ObjectMapper;

import java.io.IOException;
import java.nio.ByteBuffer;
import java.util.*;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.TimeUnit;
import java.util.stream.Collectors;

/**
 * Created by jinsheng on 16/9/18.
 *
 */
public class UpdateMppThread1 implements Runnable {

    static org.apache.log4j.Logger logger = null;

    static {

        logger = org.apache.log4j.Logger.getLogger(UpdateMppThread1.class.getName());

    }

    private String tableName;
    private String timestampKey;
    private String timestampFieldKey;
    private String filterHasField;
//    private String filterHasNoField;
    private Boolean isUpdate;
    private List<String> updateFields;
    private String updatePrimaryKey;

    private List<ColumnMetadata> columns;
    private Map<String, Type> columnMap = new HashMap<>();
    private final ObjectMapper objectMapper = new ObjectMapper();

    private Cluster cluster;
    private Session session;

    private PreparedStatement insert_statement = null;

//    @SuppressWarnings("unused")
    public void init(String keyspace,
            String dataCenter,
            String coordinatorAddress,
            String userName,
            String password,//cassandra
            Integer readTimeOut,//12000
            Boolean retry) {//true
        try {

            this.tableName = GlobalParameters.TABLE_NAME;
            keyspace = "data_wxq";
            dataCenter = "data-wxq";
            coordinatorAddress = GlobalParameters.MPP_PORT;
//        this.timestampKey = timestampKey;
//        this.timestampFieldKey = timestampFieldKey;
//        this.filterHasField = filterHasField;
//        this.filterHasNoField = filterHasNoField;
//        this.isUpdate = isUpdate;
            this.updateFields = new ArrayList<>();
            String[] lists = new String[]{"type", "cont", "uwxid", "usip", "ulc", "ulp", "mmid", "mmf", "mdf", "mccode", "murl"};
            for (String name : lists) {
                this.updateFields.add(name);
            }
//        this.updatePrimaryKey = updatePrimaryKey;
            if (readTimeOut == null) {
                readTimeOut = 0;
            }
            retry = true;
            userName = "cassandra";
            password = "cassandra";
            //创建连接
            QueryOptions options = new QueryOptions();
            logger.info("xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx");
            options.setConsistencyLevel(ConsistencyLevel.LOCAL_ONE);
            logger.info("22222222222222222222222222222222222222222222222");
            Cluster.Builder build = Cluster.builder();
            logger.info("333333333333333333333333333333333333333333");
            if (userName.length() > 0) //需要用户认证
            {
                build
                        .withSocketOptions(new SocketOptions().setConnectTimeoutMillis(120000).setReadTimeoutMillis(readTimeOut))
                        .withLoadBalancingPolicy(DCAwareRoundRobinPolicy.builder().withLocalDc(dataCenter).build())
                        //.withRetryPolicy(DefaultRetryPolicy.INSTANCE)
                        .withCredentials(userName, password).addContactPoint(coordinatorAddress).withQueryOptions(options);
//                build.withSocketOptions(new SocketOptions().setConnectTimeoutMillis(10000).setReadTimeoutMillis(0))
//					.withLoadBalancingPolicy(DCAwareRoundRobinPolicy.builder().withLocalDc("data").build())
//					.withCredentials(userName, password).addContactPoint(coordinatorAddress).withPort(port).withQueryOptions(options);
            } else // 不需要用户认证
            {
                build
                        .withSocketOptions(new SocketOptions().setConnectTimeoutMillis(120000).setReadTimeoutMillis(readTimeOut))
                        .withLoadBalancingPolicy(DCAwareRoundRobinPolicy.builder().withLocalDc(dataCenter).build())
                        //.withRetryPolicy(DefaultRetryPolicy.INSTANCE)
                        .addContactPoint(coordinatorAddress).withQueryOptions(options);
            }
            //设置重试策略
            if (retry == null || retry) {
                this.cluster = build.withRetryPolicy(DefaultRetryPolicy.INSTANCE).build();
            } else {
                this.cluster = build.build();
            }
            //绑定keyspace
            this.session = this.cluster.connect(keyspace);
            //获取该表的元数据
            this.columns = this.cluster.getMetadata().getKeyspace(keyspace).getTable(this.tableName).getColumns();

            //处理每一列的类型并保存，用于更新时确定调用哪个set方法
            columnMap = columns.parallelStream()
                    .collect(Collectors.toMap(ColumnMetadata::getName, col -> resolveType(col.getType())));
        } catch (Exception e) {
            logger.error(e, e);
        }
    }

    /**
     * 映射Cassandra类型到{@linkplain Type}枚举
     *
     * @param rawType Cassandra类型对象
     * @return 返回Cassandra类型对应的Type枚举
     */
    private Type resolveType(DataType rawType) {
        switch (rawType.getName()) {
            case ASCII:
            case VARCHAR:
            case TEXT:
                return Type.STRING;
            case TINYINT:
                return Type.TINYINT;
            case SMALLINT:
                return Type.SMALLINT;
            case INT:
                return Type.INT;
            case BIGINT:
            case VARINT:
                return Type.BIGINT;
            case FLOAT:
                return Type.FLOAT;
            case DOUBLE:
                return Type.DOUBLE;
            case BLOB:
                return Type.BLOB;
            case BOOLEAN:
                return Type.BOOLEAN;
            case LIST:
                return Type.LIST;
            case MAP:
                return Type.MAP;
            case SET:
                return Type.SET;
            default:
                throw new IllegalArgumentException("Unsupported type: " + rawType);
        }
    }

    public Session getSession() {
        return this.session;
    }

    /**
     * 设置更新条件
     *
     * @return
     */
    private String getUpdateStr() {
        StringBuilder sb = new StringBuilder();
        // 遍历需要更新的各个字段并写入where条件中
        for (String updateField : updateFields) {
            sb.append(updateField)
                    .append("=:")
                    .append(updateField)
                    .append(",");
        }
        sb.deleteCharAt(sb.length() - 1);
        return sb.toString();
    }

    /**
     * 设置更新列的值
     *
     * @param map 从消息队列中读取的数据
     * @param bs
     */
    private void setUpdateValue(Map map, BoundStatement bs) {
        // 遍历需要更新的各个列并分别设置更新值
        for (String updateField : updateFields) {
            Type type = columnMap.get(updateField);
            Object val = map.get(updateField);
            type.bind(updateField, val, bs);
        }
    }

    /**
     * 发送数据，传入参数是 List&lt;Map&gt; 类型的
     *
     * @param dataLists data list
     */
    public void pushData(List<Map<String, Object>> dataLists) throws IOException {

        if (isUpdate) {
            updateData(dataLists);
        } else {
            push_single_json_data(dataLists);
        }
    }

    /**
     * 提交数据到Cassandra
     *
     * @param jsonList 数据列表
     * @return 返回是否提交成功
     * @throws IOException 抛出异常
     */
    private boolean push_single_json_data(List<Map<String, Object>> jsonList) throws IOException {
        if (insert_statement == null) {
            // 预编译insert语句
            insert_statement = session.prepare("INSERT INTO " + this.tableName + " JSON :json if not exists");
        }

        BoundStatement bs;
        //int dropCount = 0;
        List<ResultSetFuture> future = new ArrayList<>();
        for (Object str : jsonList) {

            // 绑定数据并创建BoundStatement对象
            String json = objectMapper.writeValueAsString(str);
            bs = insert_statement.bind(json);
            // 设置记录时间戳
            if (this.timestampKey.length() > 0 && ((Map) str).containsKey(this.timestampKey)) {
                bs.setDefaultTimestamp((Long) ((Map) str).get(this.timestampKey) * 1000000); // second to microsecond
            }            //batchStatement.add(bs);
            // 异步执行查询
            ResultSetFuture result = session.executeAsync(bs);
            future.add(result);
        }

        // 处理语句执行结果
        for (int i = 0; i < future.size(); i++) {
            try {
                // 获取执行结果
                ResultSet rs = future.get(i).get();
                String result_string = rs.toString();
                System.out.println(result_string);
            } catch (OperationTimedOutException | QueryExecutionException | NoHostAvailableException e) {

                // 此处可能由于服务端原因造成，可稍后重试该操作
            } catch (InterruptedException | ExecutionException | InvalidQueryException e) {
                // 此处有可能是客户端sql语句不正确
            }
        }

        future.clear();
        return true;
    }

    /**
     * 提交更新数据到Cassandra
     *
     * @param lists 数据列表
     */
    private void updateData(List<Map<String, Object>> lists) {
//            map.remove("").remove("").remove("");
        if (insert_statement == null) {
            // 预编译update语句
            String updataSql = "update " + this.tableName + " set "
                    + getUpdateStr() + " where uchid=:uchid and mcr=:mcr and mid=:mid"
                    + " if exists";// IF EXISTS";
            insert_statement = session.prepare(updataSql);
            logger.info("udpate sql is : " + updataSql);
        }
        List<ResultSet> future = new ArrayList<>();
        logger.info("lists : " + lists.size());
        for (Map updateMap : lists) {
            logger.info("updateMap size : " + updateMap.size());
            //绑定更新数据，并创建BoundStatement对象
            BoundStatement bs = insert_statement.bind();
            setUpdateValue(updateMap, bs);
            // 绑定更新记录的主键
//            logger.info("uchidType: " + columnMap.get("uchid") + " mcrType: " + columnMap.get("mcr") + " midType: " + columnMap.get("mid"));
//            logger.info("columnMapcolumnMapcolumnMap: " + columnMap.toString());
//            logger.info("zhujian=======: "+updateMap.get("uchid").toString()+"/"+updateMap.get("mcr").toString()+"/"+updateMap.get("mid").toString());
            columnMap.get("uchid").bind("uchid", Long.valueOf(updateMap.get("uchid").toString()), bs);
            columnMap.get("mcr").bind("mcr", Long.valueOf(updateMap.get("mcr").toString()), bs);
            columnMap.get("mid").bind("mid", updateMap.get("mid").toString(), bs);
            // 设置记录时间戳
//                        if (this.timestampKey.length() > 0 && updateMap.containsKey(this.timestampKey)) {
//                            bs.setDefaultTimestamp((Long) updateMap.get(this.timestampKey) * 1000000); // second to microsecond
//                        }
            future.add(session.execute(bs));
        }
        // 处理执行结果
        logger.info("future size : " + future.size());
        for (int i = 0; i < future.size(); i++) {
//            try {
            ResultSet rs = future.get(i);
//                logger.info("ResultSet  : " + rs.toString());
            String result_string = rs.toString();
            if (result_string.contains("ERROR")) {
                logger.error(result_string);
            } else if (rs.wasApplied()) {
                //更新成功
                logger.info("update success ");
            } else {
                // TODO: 2017-03-25 此处可能是由于记录不存在等原因造成更新未能生效 by linchan
                logger.info("update unkuow error  " + result_string);
            }
//
//            } catch (Exception e) {
//                logger.error(e, e);
//                // 异常捕捉同230~234行
//            }
        }
        future.clear();

    }

    public void close() {
        this.session.close();
        this.cluster.close();
    }

    public List<ColumnMetadata> getColumns() {
        return columns;
    }

    private static <T> String arrayFormat(T[] list) {
        if (list == null || list.length == 0) {
            return null;
        }
        if (list[0].getClass() == String.class) {
            String res = "[";
            for (int i = 0; i < list.length; ++i) {
                if (i > 0) {
                    res += ",";
                }
                res += "'" + list[i].toString() + "'";
            }
            return res + "]";
        } else if (list[0].getClass() == int.class || list[0].getClass() == long.class
                || list[0].getClass() == Integer.class || list[0].getClass() == Long.class) {
            String res = "[";
            for (int i = 0; i < list.length; ++i) {
                if (i > 0) {
                    res += ",";
                }
                res += list[i].toString();
            }
            return res + "]";
        }
        return null;
    }

    private static <T> String listFormat(List<T> list) {
        return arrayFormat(list.toArray());
    }

    @Override
    public void run() {
        logger.info("demo is running running running running running running running running running ");
        try {
            init("", "", "", "", "", 0, true);
        } catch (Exception e) {
            logger.error(e, e);
        }

        while (true) {

            List<Map<String, Object>> jsonList = null;
            try {
                jsonList = new ArrayList<>();
                for (int i = 0; i < 10; ++i) {
//                    logger.info("GlobalParameters.updateQueue.size()" + GlobalParameters.mppQueue.size());
                    Map str = GlobalParameters.mppQueue.poll(500, TimeUnit.MILLISECONDS);
                    if (null != str) {
                        try {
                            //                                               1int      2int  3int     4int     5int      6long    7string  8string   9string 10string  11list
                            String updateSql = "update tp_wxq_entire_v2 set type=%s, ulc=%s, ulp=%s, mdf=%s, mccode=%s, mmid=%s, cont=%s, uwxid=%s, usip=%s, murl=%s, mmf=%s"
                                    + " where uchid=%s and mcr=%s and mid='%s'"
                                    + " if exists";// IF EXISTS";
                            List<String> list = new ArrayList<>();
                            list.add(str.get("mmf").toString().substring(1, str.get("mmf").toString().length() - 1));
                            updateSql = String.format(updateSql, str.get("type"), str.get("ulc"), str.get("ulp"), str.get("mdf"), str.get("mccode"), str.get("mmid"), str.get("cont"), str.get("uwxid"), str.get("usip"), str.get("murl"), listFormat(list),
                                    str.get("uchid"), str.get("mcr"), str.get("mid"));
                            session.execute(updateSql);
                            session.close();
                            cluster.close();
                        } catch (Exception e) {
                            logger.error(e, e);
                            if ("Timed out waiting for server response".equals(e.getMessage())) {
                                
                            } else {
                                logger.error(e, e);
                            }
                        }
                    }
                }
                if (jsonList.size() == 0) {
                    continue;
                }
                try {
                    updateData(jsonList);
                } catch (Exception e) {
                    logger.error(e, e);
                    if ("Timed out waiting for server response".equals(e.getMessage())) {
                        int num = 1;
                        while (true) {
                            logger.info("time out retry num: " + num);
                            try {
                                updateData(jsonList);
                                break;
                            } catch (Exception ex) {
                                logger.error(ex, ex);
                            }
                            if (num >= 20) {
                                close();
                                init("", "", "", "", "", 0, true);
                            }
                            if (num >= 30) {
                                logger.error("error content: " + jsonList.toString() + e, e);
                                break;
                            }
                            num++;
                            Thread.sleep(3000);

                        }
                    } else {
                        logger.error(e, e);
                    }
                }
            } catch (Exception e) {
                logger.error(e, e);
            }
        }

    }

    private enum Type {
        STRING {
            @Override
            public void bind(String col, Object val, BoundStatement bs) {
                if (val == null) {
                    bs.setToNull(col);
                } else {
                    bs.setString(col, val.toString());
                }
            }
        }, TINYINT {
            @Override
            public void bind(String col, Object val, BoundStatement bs) {
                if (val == null) {
                    bs.setToNull(col);
                } else if (val instanceof String) {
                    bs.setByte(col, Long.valueOf(((String) val).toLowerCase()).byteValue());
                } else if (val instanceof Number) {
                    bs.setByte(col, ((Number) val).byteValue());
                } else {
                    throw new IllegalArgumentException("wrong type of " + col);
                }
            }
        }, SMALLINT {
            @Override
            public void bind(String col, Object val, BoundStatement bs) {
                if (val == null) {
                    bs.setToNull(col);
                } else if (val instanceof String) {
                    bs.setShort(col, Long.valueOf(((String) val).toLowerCase()).shortValue());
                } else if (val instanceof Number) {
                    bs.setShort(col, ((Number) val).shortValue());
                } else {
                    throw new IllegalArgumentException("wrong type of " + col);
                }
            }
        }, INT {
            @Override
            public void bind(String col, Object val, BoundStatement bs) {
                if (val == null) {
                    bs.setToNull(col);
                } else if (val instanceof String) {
                    bs.setInt(col, Long.valueOf(((String) val).toLowerCase()).intValue());
                } else if (val instanceof Number) {
                    bs.setInt(col, ((Number) val).intValue());
                } else {
                    throw new IllegalArgumentException("wrong type of " + col);
                }
            }
        }, BIGINT {
            @Override
            public void bind(String col, Object val, BoundStatement bs) {
                if (val == null) {
                    bs.setToNull(col);
                } else if (val instanceof String) {
                    bs.setLong(col, Long.valueOf(((String) val).toLowerCase()));
                } else if (val instanceof Number) {
                    bs.setLong(col, ((Number) val).longValue());
                } else {
                    throw new IllegalArgumentException("wrong type of " + col);
                }
            }
        }, FLOAT {
            @Override
            public void bind(String col, Object val, BoundStatement bs) {
                if (val == null) {
                    bs.setToNull(col);
                } else if (val instanceof String) {
                    bs.setFloat(col, Long.valueOf(((String) val).toLowerCase()).floatValue());
                } else if (val instanceof Number) {
                    bs.setFloat(col, ((Number) val).floatValue());
                } else {
                    throw new IllegalArgumentException("wrong type of " + col);
                }
            }
        }, DOUBLE {
            @Override
            public void bind(String col, Object val, BoundStatement bs) {
                if (val == null) {
                    bs.setToNull(col);
                } else if (val instanceof String) {
                    bs.setDouble(col, Long.valueOf(((String) val).toLowerCase()).doubleValue());
                } else if (val instanceof Number) {
                    bs.setDouble(col, ((Number) val).doubleValue());
                } else {
                    throw new IllegalArgumentException("wrong type of " + col);
                }
            }
        }, BLOB {
            @Override
            public void bind(String col, Object val, BoundStatement bs) {
                if (val == null) {
                    bs.setToNull(col);
                } else if (val instanceof String) {
                    try {
                        byte[] bytes = Hex.decodeHex(((String) val).toCharArray());
                        bs.setBytes(col, ByteBuffer.wrap(bytes));
                    } catch (DecoderException e) {
                        throw new IllegalArgumentException("wrong value of hex string, column: " + col);
                    }
                } else if (val instanceof byte[]) {
                    bs.setBytes(col, ByteBuffer.wrap((byte[]) val));
                } else if (val instanceof ByteBuffer) {
                    bs.setBytes(col, (ByteBuffer) val);
                } else {
                    throw new IllegalArgumentException("wrong type of " + col);
                }
            }
        }, BOOLEAN {
            @Override
            public void bind(String col, Object val, BoundStatement bs) {
                if (val == null) {
                    bs.setToNull(col);
                } else if (val instanceof String) {
                    bs.setBool(col, Boolean.valueOf((String) val));
                } else if (val instanceof Boolean) {
                    bs.setBool(col, (Boolean) val);
                } else {
                    throw new IllegalArgumentException("wrong type of " + col);
                }
            }
        }, LIST {
            @Override
            public void bind(String col, Object val, BoundStatement bs) {
                if (val == null) {
                    bs.setToNull(col);
                } else if (val instanceof List) {
                    bs.setList(col, (List<?>) val);
                } else if (val instanceof Collection) {
                    List<?> l = new ArrayList<>((Collection<?>) val);
                    bs.setList(col, l);
                } else {
                    List<Object> l = new ArrayList<>();
                    l.add(val);
                    bs.setList(col, l);
                }
            }
        }, MAP {
            @Override
            public void bind(String col, Object val, BoundStatement bs) {
                if (val == null) {
                    bs.setToNull(col);
                } else if (val instanceof Map) {
                    bs.setMap(col, (Map<?, ?>) val);
                } else {
                    throw new IllegalArgumentException("wrong type of " + col);
                }
            }
        }, SET {
            @Override
            public void bind(String col, Object val, BoundStatement bs) {
                if (val == null) {
                    bs.setToNull(col);
                } else if (val instanceof Set) {
                    bs.setSet(col, (Set<?>) val);
                } else if (val instanceof Collection) {
                    Set<?> s = new HashSet<>((Collection<?>) val);
                    bs.setSet(col, s);
                } else {
                    Set<Object> s = new HashSet<>();
                    s.add(val);
                    bs.setSet(col, s);
                }
            }
        };

        public void bind(String col, Object val, BoundStatement bs) {
            throw new IllegalArgumentException("wrong type of " + col);
        }
    }
}
